package com.zyf.project.kvdb.wal.logger;

import com.zyf.project.kvdb.wal.config.WalConfig;
import com.zyf.project.kvdb.wal.walinf.Wal;
import com.zyf.project.kvdb.wal.walinf.WalDataCoder;
import com.zyf.project.kvdb.wal.walinf.WalIterator;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import utils.binary.BytesUtils;

import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * <h1>WAL 文件日志实现</h1>
 * @author Zhou Yifan
 */
@Slf4j
public class FileLogger<E extends Serializable> implements Wal<E> {

    /* 这里简单说明一下存储的结构，这个内容很重要 */
    /* |日志头信息| |(数据头信息)数据(数据头信息)| |(数据头信息)数据(数据头信息)| ... */
    /* 日志头信息 8 字节，类型为 long；数据头信息 4 字节，类型为int */

    /**
     * 日志头信息长度，目前仅用于存储快照最新位置
     */
    public static final int HEADER_SIZE = 8;

    /**
     * 日志数据头信息长度，注意和日志头信息长度不一样，目前存储数据大小
     */
    public static final int DATA_HEADER_SIZE = 4;

    /**
     * 最新快照位置，其实也就是文件通道大小，之所以称为位置是有更新和未更新的区别
     */
    private long checkpoint = 0;

    /**
     * 日志文件路径
     */
    private Path walPath;

    /**
     * 日志文件通道
     */
    private FileChannel fileChannel;

    /**
     * 日志配置项
     */
    private WalConfig config;

    /**
     * 日志数据编码器
     */
    private WalDataCoder<E> dataCoder;

    /**
     * 定时刷盘线程
     */
    private ScheduledExecutorService scheduledExecutorService;

    /**
     * 构造函数，初始化一些重要参数
     *
     * @param config    配置
     * @param dataCoder 数据编码器
     * @throws IOException IO 异常
     */
    public FileLogger(@NotNull WalConfig config, WalDataCoder<E> dataCoder) throws IOException {
        this.config = config;
        this.dataCoder = dataCoder;
        this.walPath = Paths.get(config.getLogFile());
        if (config.isTruncateExisting()) {
            this.fileChannel = FileChannel.open(walPath, StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING, StandardOpenOption.READ, StandardOpenOption.WRITE);
        } else {
            this.fileChannel = FileChannel.open(walPath, StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE);
        }
        // 初始化快照信息
        initCheckpoint();
        fileChannel.position(size());
        if (config.getFlushInterval() > 0) {
            // 设置定时刷盘的任务
            scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
            scheduledExecutorService.scheduleAtFixedRate(() -> {
                try {
                    flush();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }, 0, config.getFlushInterval(), TimeUnit.SECONDS);
        }
    }

    /**
     * 初始化快照位置
     *
     * @throws IOException IO 异常
     */
    private void initCheckpoint() throws IOException {
        long size = size();
        if (size == 0) {
            checkpoint();
        } else {
            long point = getCheckpoint();
            checkpoint = Math.min(point, size);
        }
    }

    /**
     * 获取通道大小
     *
     * @return long 通道大小，单位：byte
     * @throws IOException IO异常
     */
    public long size() throws IOException {
        return fileChannel.size();
    }

    /**
     * 追加日志
     *
     * @param data 待处理数据
     * @return 写入的字节长度
     * @throws IOException IO异常
     */
    @Override
    public synchronized int append(E data) throws IOException {
        byte[] dataBytes = dataCoder.encode(data);
        // 文件头信息的内容是 data 字节码的长度（二进制数组形式），长度是 4 字节（int 的大小）
        byte[] dataHeader = BytesUtils.toBytes(dataBytes.length);
        // header 在首尾两端所以header.length需要乘2
        byte[] bytes = new byte[dataBytes.length + dataHeader.length * 2];
        // 将 header 写到 bytes 的首尾两端
        System.arraycopy(dataHeader, 0, bytes, 0, dataHeader.length);
        System.arraycopy(dataBytes, 0, bytes, dataHeader.length, dataBytes.length);
        System.arraycopy(dataHeader, 0, bytes, dataHeader.length + dataBytes.length, dataHeader.length);
        // 写入 file 通道
        int length = fileChannel.write(ByteBuffer.wrap(bytes));
        if (config.getFlushInterval() == 0) {
            // 实时刷盘
            flush();
        }
        return length;
    }

    /**
     * 设置最新日志快照，旧的日志将不再读取，并且将在未来某一时刻被清除
     *
     * @throws IOException IO异常
     */
    @Override
    public synchronized void checkpoint() throws IOException {
        long size = size();
        if (size == 0) {
            size = HEADER_SIZE;
        }
        if (size >= config.getMaxCheckpointSize()) {
            // 如果超过阈值了，就清理快照，默认 100M
            clear();
        } else {
            // 生成新的快照，从位置0开始，位置的单位是 byte
            fileChannel.position(0);
            // 将 size 的字节码写入通道，或者说更新通道最前面的数据，长度 8 字节（快照位置是 long 类型）
            fileChannel.write(ByteBuffer.wrap(BytesUtils.toBytes(size)));
            // 位置为当前通道的大小（单位：字节）
            checkpoint = size;
            // 将当前通道位置设置在最新快照位置
            fileChannel.position(checkpoint);
        }
        if (config.getFlushInterval() == 0) {
            // 如果间隔位0，那么就是实时刷盘
            flush();
        }
    }

    public long getCheckpoint() throws IOException {
        // 将通道内的文件映射到虚拟内存，从位置0开始，映射大小为
        MappedByteBuffer byteBuffer = fileChannel.map(FileChannel.MapMode.READ_ONLY, 0, HEADER_SIZE);
        try {
            // HEADER_SIZE = 8
            byte[] bytes = new byte[HEADER_SIZE];
            // get 的参数是目标数组，这个方法将从磁盘中读取数据传递到 bytes 数组中
            byteBuffer.get(bytes);
            return BytesUtils.toLong(bytes);
        } finally {
            byteBuffer.clear();
        }
    }

    /**
     * 刷盘
     *
     * @throws IOException IO异常
     */
    @Override
    public void flush() throws IOException {
        // fileChannel.force 方法会将 Channel 里面还未写入的数据全部刷新到磁盘
        // 参数代表是否将文件元数据（比如权限等）也刷新到磁盘
        fileChannel.force(config.isFlushMetaData());
    }

    /**
     * 清空日志
     *
     * @throws IOException IO异常
     */
    @Override
    public void clear() throws IOException {
        // 相当于重新初始化
        fileChannel = FileChannel.open(walPath, StandardOpenOption.CREATE, StandardOpenOption.TRUNCATE_EXISTING, StandardOpenOption.READ, StandardOpenOption.WRITE);
        initCheckpoint();
    }


    /**
     * 前向迭代器
     *
     * @return {@link WalIterator}, impl : {@link ForwardIterator}
     */
    @Override
    public WalIterator<E> forwardIterator() {
        return new ForwardIterator();
    }

    /**
     * 反向迭代器
     *
     * @return {@link WalIterator}, impl : {@link BackwardIterator}
     */
    @Override
    public WalIterator<E> backwardIterator() throws IOException {
        return new BackwardIterator();
    }

    /**
     * 关闭当前资源，释放相关内存
     *
     * @throws Exception if this resource cannot be closed
     */
    @Override
    public void close() throws IOException {
        // 关闭事务
        if (null != scheduledExecutorService) {
            scheduledExecutorService.shutdown();
        }
        // 关闭通道
        if (null != fileChannel) {
            fileChannel.close();
        }
    }

    /**
     * 根据指定位置和大小（一般指文件头大小）获取 {@link FileLogger#fileChannel} 中的数据
     *
     * @param position 指定位置
     * @param size     文件头大小
     * @return 获取到的数据
     * @throws IOException IO 异常
     */
    private byte @Nullable [] get(long position, int size) throws IOException {
        if (position < checkpoint || size <= 0 || position > size() || position + size > size()) {
            return null;
        }
        MappedByteBuffer byteBuffer = fileChannel.map(FileChannel.MapMode.READ_ONLY, position, size);
        try {
            byte[] bytes = new byte[size];
            byteBuffer.get(bytes);
            return bytes;
        } finally {
            byteBuffer.clear();
        }
    }

    /**
     * 前向迭代器
     */
    @Getter
    class ForwardIterator implements WalIterator<E> {
        private long position;

        public ForwardIterator() {
            this.position = checkpoint;
        }

        /**
         * 判断是否有下一个
         *
         * @return 布尔值
         * @throws IOException IO异常
         */
        @Override
        public boolean hasNext() throws IOException {
            return position < size();
        }

        /**
         * 迭代到下一个
         *
         * @return 下一个数据
         * @throws IOException IO异常
         */
        @Override
        public E next() throws IOException {
            byte[] data = null;
            byte[] header = get(position, DATA_HEADER_SIZE);
            if (null != header) {
                int size = BytesUtils.toInt(header);
                data = get(position + DATA_HEADER_SIZE, size);
                position = position + DATA_HEADER_SIZE * 2 + size;
            }
            if (data == null) {
                throw new IOException("data is null");
            }
            return dataCoder.decode(data);
        }
    }

    /**
     * 反向迭代器
     */
    @Getter
    class BackwardIterator implements WalIterator<E> {
        private long position;

        public BackwardIterator() throws IOException {
            this.position = size();
        }

        /**
         * 判断是否有下一个
         *
         * @return 布尔值
         */
        @Override
        public boolean hasNext() {
            return position > checkpoint;
        }

        /**
         * 迭代到下一个
         *
         * @return 下一个数据
         * @throws IOException IO异常
         */
        @Override
        public E next() throws IOException {
            byte[] data = null;
            byte[] header = get(position - DATA_HEADER_SIZE, DATA_HEADER_SIZE);
            if (null != header) {
                int size = BytesUtils.toInt(header);
                data = get(position - DATA_HEADER_SIZE - size, size);
                position = position - DATA_HEADER_SIZE * 2 - size;
            }
            if (data == null) {
                throw new IOException("data is null");
            }
            return dataCoder.decode(data);
        }
    }

    @Override
    public String toString() {
        return "FileLogger{" +
                "checkpoint=" + checkpoint +
                ", walPath=" + walPath +
                ", config=" + config +
                '}';
    }
}
